package net.ustc.javamemcached.memorystore;

import java.io.IOException;
import java.util.Set;
import java.util.Map.Entry;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.jboss.netty.buffer.ChannelBuffer;

import net.ustc.javamemcached.Key;
import net.ustc.javamemcached.Value;
import net.ustc.javamemcached.util.OpenBitSet;

public class MemoryBucketManagerImpl implements IMemoryManager{
	/** byte number of one bucket */
	private long bucketSize;
	/** byte number of one block */
	private int blockSize;
	/** byte has been used, is multiple of blockSize */
	private long usedSize;
	/** blocks has been used */
	private int usedBlockNum;
	
	/** the cache memory is divided into many bucket */
	private ChannelBuffer bucket;
	/** used to find the idle blocks and find blocks by key */
	private MemoryLocation location;
	/** every bucket has a ReentrantReadWriteLock*/
	private ReentrantReadWriteLock storageLock = new ReentrantReadWriteLock();
	
	public MemoryBucketManagerImpl(long bucketSize, long blockSize){
		this.bucketSize = bucketSize;
		location = new MemoryLocation(bucketSize/blockSize);
	}

	public MemoryEntry get(Key key) {
		int beginIndex = location.findKeyValueLocation(key);
		if(beginIndex != -1){
			return readEntry(beginIndex);
		} else {
			return null;
		}
	}
	
	public boolean set(Key key, MemoryEntry entry) {
		remove(key);
		int entrySize = entry.getEntrySize();
		int desiredBlockNum = getDesiredBlockNum(entrySize);
		int beginBlock = location.findIdleLocation(desiredBlockNum);
		if(beginBlock != -1){
			writeEntry(entry, beginBlock);
			usedSize += entrySize;
			usedBlockNum += desiredBlockNum;
			return true;
		}else{
			return false;
		}
	}

	public boolean add(Key key, MemoryEntry entry) {
		if(contain(key)){
			return false;
		} else {
			return set(key, entry);
		}
	}
	
	public boolean contain(Key key) {
		return location.isKeyExist(key);
	}
	
	public boolean remove(Key key) {
		if(contain(key)){
			location.removeKeyValue(key);
			
			int beginBlock = location.findKeyValueLocation(key);
			int entrySize = getEntrySize(beginBlock*blockSize);
			int desiredBlockNum = getDesiredBlockNum(entrySize);
			location.reset(beginBlock, desiredBlockNum);
			usedSize -= entrySize;
			usedBlockNum -= desiredBlockNum;
			return true;
		}else {
			return false;
		}
	}
	
	public Set<Key> keySet() {
		return location.getKeySet();
	}
	
	public Set<MemoryEntry> values() {
		// TODO Auto-generated method stub
		return null;
	}	
	
	public Set<Entry<Key, MemoryEntry>> entrySet() {
		// TODO Auto-generated method stub
		return null;
	}
	
	public void clear() {
		location.close();
		usedSize = 0;
	}

	public void close() {
		this.clear();
		usedSize = 0;
	}

	public long getMemoryCapacity() {
		return bucketSize;
	}
	
	public long getUsedSize() {
		return usedSize;
	}

	private MemoryEntry readEntry(int blockIndex){
		int byteIndex = blockIndex*blockSize;
		int entrySize = getEntrySize(byteIndex);
		bucket.readerIndex(byteIndex);
		return new MemoryEntry(bucket.readSlice(entrySize));
	}
	
	private void writeEntry(MemoryEntry entry, int blockIndex){
		int byteIndex = blockIndex*blockSize;
		bucket.writerIndex(byteIndex);
		bucket.writeBytes(entry.toChannelBuffer());
	}
	
	private int getDesiredBlockNum(int entrySize) {
		return (entrySize - 1) / blockSize + 1;
	}
	
	private int getEntrySize(int byteIndex){
		bucket.readerIndex(byteIndex);
		int entrySize = bucket.readInt();
		return entrySize;
	}
	
}
